/*
 * Copyright ©2021 su binglun(9085309@qq.com). All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *---------------------------------------------------------------------
 *-                                                                   -
 *-           Website : http://phoenix.sunny360.cn/                   -
 *-                                                                   -
 *---------------------------------------------------------------------
 */
package cn.sunny360.core.wrapper.impl;

import cn.sunny360.core.common.MyProperty;
import cn.sunny360.core.wrapper.DBWrapper;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;
import io.vertx.core.json.JsonObject;
import io.vertx.mysqlclient.MySQLConnectOptions;
import io.vertx.mysqlclient.MySQLPool;
import io.vertx.sqlclient.*;

import java.util.*;

public class MysqlClientWrapper implements DBWrapper {

    private static Logger logger = LoggerFactory.getLogger(MysqlClientWrapper.class);

    private MySQLPool client;

    public MysqlClientWrapper(Vertx vertx) {
        init(vertx);
    }

    private void init(Vertx vertx) {


        MySQLConnectOptions connectOptions = new MySQLConnectOptions()
                .setPort( MyProperty.MYSQL_PORT)
                .setHost( MyProperty.MYSQL_HOST)
                .setDatabase(MyProperty.MYSQL_DATABASE)
                .setUser(MyProperty.MYSQL_USER)
                .setPassword(MyProperty.MYSQL_PASSWORD);
        // Pool options
        PoolOptions poolOptions = new PoolOptions()
                .setMaxSize(MyProperty.MYSQL_POOL_SIZE);
        // Create the client pool
        client = MySQLPool.pool(vertx, connectOptions, poolOptions);

    }


    @Override
    public Promise<Optional<JsonObject>> findOne(String sql, Tuple param) {
        Promise<Optional<JsonObject>> promise = Promise.promise();
        client.preparedQuery(sql)
                .execute(param, result->{
                    if (result.succeeded()) {
                        RowSet<Row> rows = result.result();
                        if (rows.size() == 0) {
                            promise.complete(Optional.empty());
                        } else {
                            for (Row row : rows) {
                                promise.complete(Optional.of(row.toJson()));
                                break;
                            }
                        }
                    } else {
                        promise.fail(result.cause());
                    }
                });
        return promise;
    }

    @Override
    public Promise<List<JsonObject>> findList(String sql, Tuple param) {
        Promise<List<JsonObject>> promise = Promise.promise();
        client.preparedQuery(sql)
                .execute(param, result->{
                    if (result.succeeded()) {
                        RowSet<Row> rows = result.result();
                        List<JsonObject> list = new ArrayList<>();
                        for (Row row : rows) {
                            list.add(row.toJson());
                        }
                        promise.complete(list);
                    } else {
                        promise.fail(result.cause());
                    }
                });
        return promise;
    }

    @Override
    public Promise<Integer> insert(String sql, Tuple param) {
        return execute(sql, param);
    }

    @Override
    public Promise<Integer> update(String sql, Tuple param) {
        return execute(sql, param);
    }

    @Override
    public Promise<Integer> delete(String sql, Tuple param) {
        return execute(sql, param);
    }

    private Promise<Integer> execute(String sql, Tuple param) {
        Promise<Integer> promise = Promise.promise();
        client.preparedQuery(sql)
                .execute(param, result->{
                    if (result.failed()) {
                        promise.fail(result.cause());
                    } else {
                        RowSet<Row> rows = result.result();
                        promise.complete(result.result().rowCount());
                    }
                });
        return promise;
    }


    @Override
    public Promise<Boolean> executeWithTrans(List<SQLTuple> sqls) {
        Promise<Boolean> promise = Promise.promise();
        client.getConnection()
                .onSuccess(conn->{
                    Future<Transaction> future = conn.begin();
                    future.compose(tx-> execSqlChain(conn, future, sqls, 0)
                                .compose(res->Future.succeededFuture(tx)))
                    .compose(tx-> tx.commit())
                            .eventually(event->conn.close())
                            .onSuccess(h->promise.complete(true))
                            .onFailure(e->{
                                Throwable t = (Throwable) e;
                                logger.error("Transaction failed!", t);
                                promise.fail(t);
                            });

                })
                .onFailure(e->{
                    logger.error("Open connection error.", e);
                    promise.fail(e);
                });
        return promise;
    }

    @Override
    public Promise<Boolean> executeWithTrans(SQLTuple... sqls) {
        return executeWithTrans(Arrays.asList(sqls));
    }

    private Future<?> execSqlChain(SqlConnection conn, Future<?> future, List<SQLTuple> sqls, int i) {
        if (i < sqls.size()) {
            SQLTuple sql = sqls.get(i);
            Future f = future
                    .compose(tx -> conn.preparedQuery(sql.getSql()).execute(sql.getParam()));
            return execSqlChain(conn, f, sqls, i+1);
        } else {
            return future;
        }
    }



}
